package cn.doitedu.sometest;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/19
 * @Desc: 学大数据，上多易教育
 *
 * 有如下数据：
 *     user_id,event_id,action_time,session_id,money
 *     1,event_a,1000,s01,180
 *     1,event_x,2000,s01,280
 *     1,event_x,3000,s01,280
 *     1,event_x,4000,s01,380
 *     1,event_w,5000,s01,420
 *     1,event_x,6000,s01,420
 *     1,event_y,7000,s01,420
 *     1,event_a,8000,s01,610
 *     1,event_b,9000,s01,530
 *     1,event_b,10000,s01,530
 *     1,event_b,11000,s01,580
 *
 * 实时计算满足如下规则的数据
 *    发生event_a后，再发生若干次 event_x,然后再发生 event_w
 *    然后，输出 event_a的 action_time, event_x的平均money和最大money，event_w的 action_time
 *
 *
 **/
public class FlinkCepDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        DataStreamSource<String> stream = env.socketTextStream("doitedu", 9988);
        SingleOutputStreamOperator<SomeBean> mapped = stream.map(line -> {
            String[] split = line.split(",");
            return new SomeBean(split[0], split[1], Long.parseLong(split[2]),
                    split[3], Float.parseFloat(split[4]));
        });

        tenv.createTemporaryView("tmp",mapped,
                Schema.newBuilder()
                        .column("user_id", DataTypes.STRING())
                        .column("event_id", DataTypes.STRING())
                        .column("action_time", DataTypes.BIGINT())
                        .column("session_id", DataTypes.STRING())
                        .column("money", DataTypes.FLOAT())
                        .columnByExpression("rt","to_timestamp_ltz(action_time,3)")
                        .watermark("rt","rt")
                        .build());

        tenv.executeSql(
                        " SELECT                                     "+
                        "   a_action_time as start_time,             "+
                        " 	avg_b_money,                             "+
                        " 	max_b_money,                             "+
                        " 	c_action_time as end_time                "+
                        " FROM tmp                                   "+
                        "     MATCH_RECOGNIZE(                       "+
                        " 	    PARTITION BY user_id,session_id      "+
                        " 		ORDER BY rt                          "+
                        " 		MEASURES                             "+
                        " 		    A.action_time as a_action_time,  "+
                        " 			avg(B.money) as avg_b_money,     "+
                        " 			max(B.money) as max_b_money,     "+
                        " 			C.action_time as c_action_time   "+
                        " 		ONE ROW PER MATCH                    "+
                        " 		AFTER MATCH SKIP PAST LAST ROW       "+
                        " 		PATTERN(A B+ C)                      "+
                        "       DEFINE                               "+
                        " 		  A AS A.event_id = 'event_a',       "+
                        " 		  B AS B.event_id = 'event_x',       "+
                        " 		  C AS C.event_id = 'event_w'	     "+
                        " 	)                                        "
        ).print();


        env.execute();


    }


    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class SomeBean{
        String user_id;
        String event_id;
        long   action_time;
        String session_id;
        float money;
    }



}
